package com.zyh.flink.day07.evictor;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.evictors.CountEvictor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.CountTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class CountEvictorTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> dataStreamSource = environment.socketTextStream("hadoop10", 9998)
                .flatMap(new FlatMapFunction<String, String>() {
                    @Override
                    public void flatMap(String value, Collector<String> out) throws Exception {
                        String[] words = value.split("\\s+");
                        for (String word : words) {
                            out.collect(word);
                        }
                    }
                });

        AllWindowedStream<String, TimeWindow> windowedStream = dataStreamSource
                .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                .trigger(CountTrigger.of(3))
                //.evictor(CountEvictor.of(5));//窗口中的元素不能超过5个
                .evictor(CountEvictor.of(0,true));//计算之后剔除,保留0个

        SingleOutputStreamOperator<String> result = windowedStream.reduce(new ReduceFunction<String>() {
            @Override
            public String reduce(String value1, String value2) throws Exception {
                return value1 + "*" + value2;
            }
        });

        result.print();

        environment.execute("countEvictor");
    }
}